/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
package cn.ac.iie.di.msg;

import cn.ac.iie.di.datadock.rdata.exchange.client.core.REFieldType;
import cn.ac.iie.di.datadock.rdata.exchange.client.exception.REConnectionException;
import cn.ac.iie.di.datadock.rdata.exchange.client.exception.RESessionException;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.connection.REConnection;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.session.RESendSession;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.session.RESendSessionBuilder;
import cn.ac.iie.di.global.GlobalParas;
import static cn.ac.iie.di.global.GlobalParas.SEND_POOL;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.Map;
import java.util.NoSuchElementException;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import org.apache.log4j.Logger;

/**
 *
 * @author lenovo-pc
 */
public class JsonMsgSender implements Runnable {

    private int addCount = 0;
    private long startTime = System.currentTimeMillis();
    static private final Logger logger = Logger.getLogger(JsonMsgSender.class.getName());
    private int pool_msg = 0;
    private String ip_port = null;
    private String mq_msg = null;
    private String type_msg = null;
    private String sendName = null;
    private ArrayBlockingQueue<Map> sendQueue = null;
    public String threadName = null;

    ThreadPoolExecutor threadpool;

    public JsonMsgSender(int pool_msgs, String ip_ports, String mq_msgs, String type_msgs, ArrayBlockingQueue<Map> sendQueues, String threadNames, String sendName) {
        this.pool_msg = pool_msgs;
        this.ip_port = ip_ports;
        this.mq_msg = mq_msgs;
        this.type_msg = type_msgs;
        this.sendQueue = sendQueues;
        this.threadName = threadNames;
        this.sendName = sendName;

        //创建一个线程池
        threadpool = new ThreadPoolExecutor(GlobalParas.SEND_POOL, GlobalParas.SEND_POOL, 3, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>(this.pool_msg), new ThreadFactory() {
//            threadpool = new ThreadPoolExecutor(12, 12, 3, TimeUnit.SECONDS,new LinkedBlockingQueue<Runnable>(20),new ThreadFactory() {
            AtomicInteger idx = new AtomicInteger(0);

            @Override
            public Thread newThread(Runnable r) {
                return new Thread(r, threadName + idx.getAndIncrement());
            }
        }, new RejectedExecutionHandler() {

            @Override
            public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                while (true) {
                    try {
                        executor.getQueue().put(r);
                        break;
                    } catch (InterruptedException ex) {
                        logger.warn("", ex);
                    }
                }
            }
        });
    }

    @Override
    public void run() {
        logger.info("JsonMsgSender is running");
        RESendSession session;
        ReentrantReadWriteLock readWriteLock = new ReentrantReadWriteLock();
        ReentrantReadWriteLock.ReadLock readlock = readWriteLock.readLock();
        ReentrantReadWriteLock.WriteLock writelock = readWriteLock.writeLock();
        SessionPool pool;
        try {
            REConnection conn = new REConnection(this.ip_port);
            // REConnection conn = new REConnection("10.136.64.28:9877");
            //builder
            RESendSessionBuilder builder = (RESendSessionBuilder) conn.getSendSessionBuilder(this.mq_msg);
            //RESendSessionBuilder builder = (RESendSessionBuilder) conn.getSendSessionBuilder("qinjia_mq");
            //群发的类型初始化
            logger.info("====++++======" + this.type_msg);
            for (String columntype : this.type_msg.split("\\|")) {
                logger.debug("config columnTypes are "+columntype);
                switch (columntype.split("[:]")[1]) {
                    case "int":
                        builder.addColumn(columntype.split("[:]")[0], REFieldType.Int, true);
                        break;
                    case "ints":
                        builder.addColumn(columntype.split("[:]")[0], REFieldType.Ints, true);
                        break;
                    case "long":
                        builder.addColumn(columntype.split("[:]")[0], REFieldType.Long, true);
                        break;
                    case "longs":
                        builder.addColumn(columntype.split("[:]")[0], REFieldType.Longs, true);
                        break;
                    case "string":
                        builder.addColumn(columntype.split("[:]")[0], REFieldType.String, true);
                        break;
                    case "strings":
                        logger.info("set strings: " + columntype.split("[:]")[0]);
                        builder.addColumn(columntype.split("[:]")[0], REFieldType.Strings, true);
                        break;
                    case "boolean":
                        builder.addColumn(columntype.split("[:]")[0], REFieldType.Boolean, true);
                        break;
                    case "struct":
                        HashMap<String, REFieldType> detailMap1 = new HashMap<>();
                        detailMap1.put("tct_l1", REFieldType.Long);
                        detailMap1.put("tct_l2", REFieldType.Long);
                        detailMap1.put("tct_l3", REFieldType.Long);
                        builder.addColumn(columntype.split("[:]")[0], REFieldType.Struct, true);
                        REFieldType.Struct.setDetail(columntype.split("[:]")[0], detailMap1);
                        break;
                    case "structs":
                        HashMap<String, REFieldType> detailMap2 = new HashMap<>();
                        detailMap2.put("v", REFieldType.String);
                        detailMap2.put("w", REFieldType.Long);
                        builder.addColumn(columntype.split("[:]")[0], REFieldType.Structs, true);
                        REFieldType.Structs.setDetail(columntype.split("[:]")[0], detailMap2);
                        break;
                    default:
                        logger.error("The type is not found");
                }
                logger.debug("configuration columnTypes are "+columntype);
            }
            pool = new SessionPool(builder);
            //session
            session = (RESendSession) builder.build();

            session.start();

        } catch (Exception ex) {
            logger.error("init session " + ex, ex);
            return;
        }

        String json = null;
        ArrayList<Map> jsonList;
        while (true) {
            jsonList = new ArrayList<>(50);
            try {

                for (int i = 0; i < 50; ++i) {
                    Map str = this.sendQueue.poll(500, TimeUnit.MILLISECONDS);
                    if (str == null) {
                        break;
                    }
                    jsonList.add(str);
                }
                if (jsonList.isEmpty()) {
                    Thread.sleep(5000);
                    continue;
                }
                threadpool.execute(new JsonMsgTask(jsonList, pool, this.sendName, this.type_msg));
                ++addCount;
            } catch (Exception ex) {
                logger.error("this group thread error :" + ex, ex);
                ex.printStackTrace();
            }
        }

    }

    static public class SessionPool {

        RESendSessionBuilder builder;
        LinkedList<RESendSession> pool;

        public SessionPool(RESendSessionBuilder builder) {
            this.builder = builder;
            pool = new LinkedList<>();
        }

        public synchronized RESendSession borrowSession() throws RESessionException, REConnectionException {
            try {
                return pool.removeFirst();
            } catch (NoSuchElementException ex) {
                RESendSession rtn = (RESendSession) builder.build();
                rtn.start();
                return rtn;
            }
        }

        public synchronized void returnSession(RESendSession session) {
            pool.add(session);
        }
    }

}
